package DataCollect;

import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import dms.MATPOuterClass;
import dms.MTransfCtrlOuterClass;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.zeromq.ZMQ;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.Properties;

public class ZmqCollectMainData {
    public static void main(String[] args) throws IOException {
        ZMQ.Context zmq = ZMQ.context(1);

        ZMQ.Socket sub = zmq.socket(ZMQ.SUB);

        sub.connect("tcp://localhost:5555");

        sub.subscribe("".getBytes());

        while (true){
            byte[] recv = sub.recv();

            //将接受的字节数组转为MTransfCtrl类
            MTransfCtrlOuterClass.MTransfCtrl mTransfCtrl1 = MTransfCtrlOuterClass.MTransfCtrl.parseFrom(recv);

            //定义kafka生产者
            Properties properties = new Properties();
            //key序列化
            properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            //value序列化
            properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            //设置服务端主机名
            properties.put("bootstrap.servers", "hadoop005:9092");

            KafkaProducer<String, String> producer = new KafkaProducer<>(properties);


            StringBuffer sb = new StringBuffer();
            //定义MPacketHead字段值
            String MPacketHead_atpType = "";
            String MPacketHead_trainID = "";
            String MPacketHead_trainNum = "";
            String MPacketHead_AttachRWBureau = "";
            String MPacketHead_ViaRWBureau = "";
            String MPacketHead_CrossDayTrainNum = "";
            String MPacketHead_DriverID = "";


            if (mTransfCtrl1.hasMsgtype() && mTransfCtrl1.getMsgtype() == 1){
                ByteString data = mTransfCtrl1.getData();

                MATPOuterClass.MATP matp = MATPOuterClass.MATP.parseFrom(data);

                if (matp.hasPacketHead() && matp.getPacketHead().hasATPType()){
                    MPacketHead_atpType = String.valueOf(matp.getPacketHead().getATPType());
                }

                if (matp.hasPacketHead() && matp.getPacketHead().hasTrainID()){
                    MPacketHead_trainID = String.valueOf(matp.getPacketHead().getTrainID());
                }
                if (matp.hasPacketHead() && matp.getPacketHead().hasTrainNum()){
                    MPacketHead_trainNum = matp.getPacketHead().getTrainNum().toStringUtf8();
                }
                if (matp.hasPacketHead() && matp.getPacketHead().hasAttachRWBureau()){
                    MPacketHead_AttachRWBureau = String.valueOf(matp.getPacketHead().getAttachRWBureau());
                }
                if (matp.hasPacketHead() && matp.getPacketHead().hasViaRWBureau()){
                    MPacketHead_ViaRWBureau = String.valueOf(matp.getPacketHead().getViaRWBureau());
                }
                if (matp.hasPacketHead() && matp.getPacketHead().hasCrossDayTrainNum()){
                    MPacketHead_CrossDayTrainNum = String.valueOf(matp.getPacketHead().getCrossDayTrainNum());
                }
                if (matp.hasPacketHead() && matp.getPacketHead().hasDriverID()){
                    MPacketHead_DriverID = matp.getPacketHead().getDriverID().toStringUtf8();
                }

                //获取uuid
                String uuid = "";
                if (matp.hasUUID()){
                    uuid = matp.getUUID().toStringUtf8();
                }
                String tmp = "";
                //获取温度
                if(matp.hasTemperature()){
                    tmp = String.valueOf(matp.getTemperature());
                }

                String road = "";
                if (matp.hasRoad()){
                    road = String.valueOf(matp.getRoad());
                }

                //获取经纬度
                String lng = "";
                if (matp.hasSignal() && matp.getSignal().hasLongitude()){
                    lng = String.valueOf(matp.getSignal().getLongitude());
                }

                String lat = "";
                if (matp.hasSignal() && matp.getSignal().hasLatitude()){
                    lat = String.valueOf(matp.getSignal().getLatitude());
                }

                String weather = "";
                if (StringUtils.isNotEmpty(lng) && StringUtils.isNotEmpty(lat)){
                    weather = MakeWeather.getWeather(lng, lat);
                }


                sb.append(MPacketHead_atpType).append("|")
                        .append(MPacketHead_trainID).append("|")
                        .append(MPacketHead_trainNum).append("|")
                        .append(MPacketHead_AttachRWBureau).append("|")
                        .append(MPacketHead_ViaRWBureau).append("|")
                        .append(MPacketHead_CrossDayTrainNum).append("|")
                        .append(MPacketHead_DriverID).append("|")
                        /*
                        .........
                         */
                        .append(uuid).append("|")
                        .append(tmp).append("|")
                        .append(road).append("|")
                        .append(weather);
                 producer.send(new ProducerRecord<String, String>("train086", sb.toString()));
            }

            System.out.println(sb.toString());


        }
    }
}
